package com.alison.sparkstream.transform

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object E1_transform {

  """
    |转换
    |
    |map
    |filter
    |flatmap
    |repartition
    |reduceByKey
    |countByValue
    |groupByKey
    |transform
    |join
    |""".stripMargin
  """
    |无状态转换操作是指那些不需要维护状态信息的转换操作，每个批次之间相互独立，不依赖于之前的数据处理结果。即把简单的 RDD 转化操作应用到每个批次上，也就是转化 DStream 中的每一个 RDD。
    |无状态转换操作非常适合于一些独立的数据转换和过滤，比如映射、过滤、扁平化等操作。
    |""".stripMargin

  def main(args: Array[String]): Unit = {
    joiner()
  }

  // 创建 Spark 运行配置对象
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("E1_transform")
  // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
  val ssc = new StreamingContext(sparkConf, Seconds(10))

  def maper() = {
    // 假设我们有一个输入DStream
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 使用map操作将每个元素转换为其长度
    val lengthDStream: DStream[Int] = inputDStream.map(_.length)
    // 打印结果
    lengthDStream.print()
  }

  def flatMaper() = {
    // 假设我们有一个输入DStream
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 使用flatMap操作将每个元素拆分成单词
    val wordsDStream: DStream[String] = inputDStream.flatMap(_.split(" "))
    // 打印结果
    wordsDStream.print()
  }

  def filterer() = {
    // 假设我们有一个输入DStream
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)

    // 将输入DStream中的字符串元素转换为整数
    val intDStream: DStream[Int] = inputDStream.map(_.toInt)

    // 使用filter操作过滤出偶数元素
    val evenDStream: DStream[Int] = intDStream.filter(_ % 2 == 0)

    // 打印结果
    evenDStream.print()
  }
  """
    |repartition 操作用于重新分区 DStream 中的数据，以便更有效地进行数据处理和并行计算，类似于 RDD 批处理中的 repartition。
    |通过重新分区，我们可以调整数据的分布，使得后续的处理可以更加均衡和高效。
    |
    |注意：repartition 操作可能会引起数据的洗牌（shuffle），因此它可能会产生一些性能开销。
    |通常情况下，我们应该在确实需要改变数据分区的情况下使用 repartition，
    |以便在并行计算和负载均衡方面获得更好的效果。
    |""".stripMargin

  def reparitioner() = {
    // 假设我们有一个输入DStream
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 使用repartition操作重新分区DStream中的数据
    val repartitionedDStream = inputDStream.repartition(4) // 假设我们要分成4个分区
    // 输出结果
    repartitionedDStream.print()
  }

  def reduceByKeyer() = {
    // 假设我们有一个输入DStream
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 将输入DStream中的元素转换为键值对数据 (word, count)
    val keyValueDStream = inputDStream.map(line => (line, 1))
    // 使用reduceByKey操作对相同键的值进行求和
    val sumByKeyDStream = keyValueDStream.reduceByKey(_ + _)
    // 打印结果
    sumByKeyDStream.print()
  }


  def countByValueer() = {
    // 假设我们有一个输入DStream
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 使用countByValue操作统计每个元素出现的次数
    val countDStream = inputDStream.countByValue()
    // 打印结果
    countDStream.print()
  }

  def groupByKeyer() = {
    // 假设我们有一个输入DStream
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 将输入DStream中的元素转换为键值对数据 (word, count)
    val keyValueDStream = inputDStream.map(line => (line, 1))
    // 使用groupByKey操作对相同键的值进行分组
    val groupedDStream = keyValueDStream.groupByKey()
    // 输出结果
    groupedDStream.print()
  }

  def transformer() = {
    // 假设我们有一个输入DStream
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)
    // 使用transform操作对每个RDD进行自定义转换，
    val transformedDStream: DStream[String] = inputDStream.transform { rdd =>
      // 将每个字符串转换为大写
      rdd.map(_.toUpperCase())
    }
    // 输出结果
    transformedDStream.print()
  }

  """
    |nc -lk 9999
    |>a,1
    |>a,2
    |>b,3
    |>c,4
    |
    |nc -lk 8888
    |>a,hello
    |>b,go
    |>c,com
    |>d,bye
    |
    |res
    |(a,(1,hello))
    |(a,(2,hello))
    |(b,(3,go))
    |""".stripMargin

  def joiner() = {
    // 创建 Spark 运行配置对象
//    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("E1_transform")
    // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
//    val ssc = new StreamingContext(sparkConf, Seconds(3))
    // 创建两个输入DStream，模拟键值对数据
    val stream1 = ssc.socketTextStream("192.168.56.104", 9999)
    val stream2 = ssc.socketTextStream("192.168.56.104", 8888)
    stream1.print()
    println("---------------")
    stream2.print()
    // 将两个流转换为 KV 类型
    val dstream1 = stream1.map(line => {
      val parts = line.split(",")
      (parts(0), parts(1))
    })
    val dstream2 = stream2.map(line => {
      val parts = line.split(",")
      (parts(0), parts(1))
    })
    // 使用join操作将具有相同键的数据合并
    val joinedDStream = dstream1.join(dstream2)

    // 输出结果
    joinedDStream.print()
    // 开启任务
    ssc.start()
    // 等待应用程序终止
    ssc.awaitTermination()
  }


}
